
#include "aco/aco-worker.h"
#include "aco/aco-socket.h"
#include "zalloc.h"
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>
#include <sys/uio.h>

#ifndef EPOLLONESHOT
#define EPOLLONESHOT 0x40000000
#endif

// AIO操作类型
typedef enum {
    AIO_OP_ACCEPT,
    AIO_OP_CONNECT,
    AIO_OP_RECV,
    AIO_OP_SEND,
    AIO_OP_READ,
    AIO_OP_WRITE,
    AIO_OP_RECVFROM,
    AIO_OP_SENDTO,
    AIO_OP_RECV_V,
    AIO_OP_SEND_V,
    AIO_OP_RECVFROM_V,
    AIO_OP_SENDTO_V,
    AIO_OP_RECVMSG,
    AIO_OP_SENDMSG,
    AIO_OP_RECVMSG_V,
    AIO_OP_SENDMSG_V
} aco_op_type_t;


// AIO操作结构体
typedef struct aco_sockopt_s {
    aco_op_type_t type;
    socket_t fd; // 直接使用文件描述符
    
    // 用户参数和回调函数
    void* param;
    union {
        // connect/accpet
        aco_onaccept   onaccept;
        aco_onconnect  onconnect;
        // send
        aco_onsend     onsend;
        aco_onsendto   onsendto;
        // recv
        aco_onrecv     onrecv;
        aco_onrecvfrom onrecvfrom;
        aco_onrecvmsg  onrecvmsg;
        
    } callback;
    
    // 操作特定数据
    union {
        struct {
            // 无额外数据
        } accept;
        
        struct {
            void* buffer;
            size_t bytes;
        } recv;

        struct {
            void* buffer;
            size_t bytes;
        } recvfrom;

        struct {
            socket_bufvec_t* vec;
            int n;
        } recv_v;
        
        struct {
            socket_bufvec_t* vec;
            int n;
        } recvfrom_v;
        
        struct {
            void* buffer;
            size_t bytes;
        } recvmsg;

        struct {
            socket_bufvec_t* vec;
            int n;
        } recvmsg_v;

    } in;

    union {
        struct {
            const struct sockaddr_storage* addr;
            socklen_t addrlen;
        } connect;
        
        struct {
            const void* buffer;
            size_t bytes;
        } send;

        struct {
            const struct sockaddr_storage* addr;
            socklen_t addrlen;
            const void* buffer;
            size_t bytes;
        } sendto;

        struct {
            socket_bufvec_t* vec;
            int n;
        } send_v;

        struct {
            const struct sockaddr_storage* addr;
            socklen_t addrlen;
            socket_bufvec_t* vec;
            int n;
        } sendto_v;

        struct {
            const struct sockaddr_storage* peer;
            socklen_t peerlen;
            const struct sockaddr_storage* local;
            socklen_t locallen;
            const void* buffer;
            size_t bytes;
        } sendmsg;

        struct {
            const struct sockaddr_storage* peer;
            socklen_t peerlen;
            const struct sockaddr_storage* local;
            socklen_t locallen;
            socket_bufvec_t* vec;
            int n;
        } sendmsg_v;
        
    } out;
    
    uint32_t events;     // epoll监听的事件类型
    uint32_t timeout_ms; // 超时时间
    struct aco_sockopt_s* next;
} aco_sockopt_t;

// 新增操作池管理结构
typedef struct {
    aco_sockopt_t* free_list;
    size_t pool_size;
    size_t max_pool_size;
} aco_sockopt_pool_t;

// 全局操作池（线程安全实现）
static __thread aco_sockopt_pool_t global_op_pool = {
    .free_list = NULL,
    .pool_size = 0,
    .max_pool_size = 1024
};

// 从池中获取操作结构体
static aco_sockopt_t* op_pool_alloc(void) 
{
    if (global_op_pool.free_list) {
        aco_sockopt_t* op = global_op_pool.free_list;
        global_op_pool.free_list = op->next;
        global_op_pool.pool_size--;
        memset(op, 0, sizeof(aco_sockopt_t)); // 安全清零
        return op;
    }
    return z_calloc(1, sizeof(aco_sockopt_t)); // 池为空则动态分配
}

// 将操作结构体归还到池中
static void op_pool_free(aco_sockopt_t* op) 
{
    if (!op) return;
    
    // 简单回收策略：池未满则回收
    if (global_op_pool.pool_size < global_op_pool.max_pool_size) { // 最大池大小限制
        op->next = global_op_pool.free_list;
        global_op_pool.free_list = op;
        global_op_pool.pool_size++;
    } else {
        z_free(op); // 池已满则直接释放
    }
}

// 初始化操作池
void aco_socket_pool_init(size_t initial_size) 
{
    global_op_pool.free_list = NULL;
    global_op_pool.pool_size = 0;
    global_op_pool.max_pool_size = initial_size > 0 ? initial_size : 1024;
    
    // 预分配操作结构体
    for (size_t i = 0; i < initial_size; i++) {
        aco_sockopt_t* op = z_calloc(1, sizeof(aco_sockopt_t));
        if (op) {
            op->next = global_op_pool.free_list;
            global_op_pool.free_list = op;
            global_op_pool.pool_size++;
        }
    }
}

// 销毁操作池
void aco_socket_pool_destroy(void) 
{
    aco_sockopt_t* current = global_op_pool.free_list;
    while (current) {
        aco_sockopt_t* next = current->next;
        z_free(current);
        current = next;
    }
    global_op_pool.free_list = NULL;
    global_op_pool.pool_size = 0;
}

// 创建AIO socket（简化版）
aco_socket_t aco_socket_create(socket_t socket) 
{
    aco_sockopt_t* op = op_pool_alloc();
    
    // 设置为非阻塞模式
    if (socket_setnonblock(socket, 1) == -1) 
    {
        perror("set nonblock:");
    }

    // 直接返回socket描述符
    op->fd = socket;
    return (aco_socket_t)(intptr_t)op;
}

// 销毁AIO socket
int aco_socket_destroy(aco_socket_t socket, aco_ondestroy ondestroy, void* param) 
{
    aco_sockopt_t* op = (aco_sockopt_t*)socket;

    // 执行shutdown确保所有pending操作完成
    shutdown(op->fd, SHUT_RDWR);
    close(op->fd);
    
    if (ondestroy) {
        ondestroy(param);
    }

    op_pool_free(op);
    return 0;
}

// 任务执行函数
static void aco_socket_entry(void* arg) 
{
    aco_task_t* task = (aco_task_t*)aco_get_arg();
    aco_sockopt_t* op = (aco_sockopt_t*)arg;
    int result = 0;
    size_t bytes = 0;

    // 检查任务是否已超时
    if (task && task->state == ACO_TASK_TIMEOUT) 
    {
        // 根据操作类型调用相应的超时回调
        switch(op->type) {
        case AIO_OP_ACCEPT:
            if (op->callback.onaccept) 
                op->callback.onaccept(op->param, ETIMEDOUT, -1, NULL, 0);
            break;
            
        case AIO_OP_CONNECT:
            if (op->callback.onconnect) 
                op->callback.onconnect(op->param, ETIMEDOUT);
            break;
            
        case AIO_OP_READ:    
        case AIO_OP_RECV:
        case AIO_OP_RECV_V:
            if (op->callback.onrecv) 
                op->callback.onrecv(op->param, ETIMEDOUT, 0);
            break;

        case AIO_OP_WRITE:
        case AIO_OP_SEND:
        case AIO_OP_SEND_V:
        case AIO_OP_SENDMSG:
        case AIO_OP_SENDMSG_V:    
            if (op->callback.onsend) 
                op->callback.onsend(op->param, ETIMEDOUT, 0);
            break;

        case AIO_OP_RECVFROM:
        case AIO_OP_RECVFROM_V:
            if (op->callback.onrecvfrom) 
                op->callback.onrecvfrom(op->param, ETIMEDOUT, 0, NULL, 0);
            break;

        case AIO_OP_SENDTO:
        case AIO_OP_SENDTO_V:
            if (op->callback.onsendto) 
                op->callback.onsendto(op->param, ETIMEDOUT, 0, NULL, 0);
            break;

        case AIO_OP_RECVMSG:
        case AIO_OP_RECVMSG_V:
            if (op->callback.onrecvmsg) 
                op->callback.onrecvmsg(op->param, ETIMEDOUT, 0, NULL, 0, NULL, 0);
            break;

        default:
            break;
        }
        
        return;
    }

    switch(op->type) {
    case AIO_OP_ACCEPT: {
            struct sockaddr_storage addr;
            socklen_t addrlen = sizeof(addr);
            socket_t new_fd = socket_accept(op->fd, &addr, &addrlen);
            
            result = (new_fd == -1) ? errno : 0;
            if (op->callback.onaccept) {
                op->callback.onaccept(op->param, result, new_fd, &addr, addrlen);
            }
            break;
        }
    
    case AIO_OP_CONNECT: {
            result = socket_connect(op->fd, (struct sockaddr*)op->out.connect.addr, 
                                    op->out.connect.addrlen);
            result = (result == -1) ? errno : 0;
            if (op->callback.onconnect) {
                op->callback.onconnect(op->param, result);
            }
            break;
        }

    case AIO_OP_READ:
    case AIO_OP_RECV: {
            ssize_t n = socket_recv(op->fd, op->in.recv.buffer, op->in.recv.bytes, 0);
            result = (n == -1) ? errno : 0;
            bytes = (n == -1) ? 0 : n;
            if (op->callback.onrecv) {
                op->callback.onrecv(op->param, result, bytes);
            }
            break;
        }
    
    case AIO_OP_WRITE:
    case AIO_OP_SEND: {
            ssize_t n = socket_send(op->fd, op->out.send.buffer, op->out.send.bytes, 0);
            result = (n == -1) ? errno : 0;
            bytes = (n == -1) ? 0 : n;
            if (op->callback.onsend) {
                op->callback.onsend(op->param, result, bytes);
            }
            break;
        }

    case AIO_OP_RECV_V: {
            ssize_t n = socket_recv_v(op->fd, op->in.recv_v.vec, op->in.recv_v.n, 0);
            result = (n == -1) ? errno : 0;
            bytes = (n == -1) ? 0 : n;
            if (op->callback.onrecv) {
                op->callback.onrecv(op->param, result, bytes);
            }
            break;
        }

    case AIO_OP_SEND_V: {
            ssize_t n = socket_send_v(op->fd, op->out.send_v.vec, op->out.send_v.n, 0);
            result = (n == -1) ? errno : 0;
            bytes = (n == -1) ? 0 : n;
            if (op->callback.onsend) {
                op->callback.onsend(op->param, result, bytes);
            }
            break;
        }
    
    case AIO_OP_RECVFROM: {
            struct sockaddr_storage addr;
            socklen_t addrlen = sizeof(addr);
            ssize_t n = socket_recvfrom(op->fd, op->in.recvfrom.buffer, op->in.recvfrom.bytes, 0, 
                                        (struct sockaddr*)&addr, &addrlen);
            result = (n == -1) ? errno : 0;
            bytes = (n == -1) ? 0 : n;
            if (op->callback.onrecvfrom) {
                op->callback.onrecvfrom(op->param, result, bytes, &addr, addrlen);
            }
            break;
        }
    
    case AIO_OP_SENDTO: {
            ssize_t n = socket_sendto(op->fd, op->out.sendto.buffer, op->out.sendto.bytes, 0, 
                                    (struct sockaddr*)op->out.sendto.addr, op->out.sendto.addrlen);
            result = (n == -1) ? errno : 0;
            bytes = (n == -1) ? 0 : n;
            if (op->callback.onsendto) {
                op->callback.onsendto(op->param, result, bytes, 
                                    op->out.sendto.addr, op->out.sendto.addrlen);
            }
            break;
        }

    case AIO_OP_RECVFROM_V: {
            struct sockaddr_storage addr;
            socklen_t addrlen = sizeof(addr);
            ssize_t n = socket_recvfrom_v(op->fd, op->in.recvfrom_v.vec, op->in.recvfrom_v.n, 0, 
                                        (struct sockaddr*)&addr, &addrlen);
            result = (n == -1) ? errno : 0;
            bytes = (n == -1) ? 0 : n;
            if (op->callback.onrecvfrom) {
                op->callback.onrecvfrom(op->param, result, bytes, &addr, addrlen);
            }
            break;
        }
        
    case AIO_OP_SENDTO_V: {
            ssize_t n = socket_sendto_v(op->fd, op->out.sendto_v.vec, op->out.sendto_v.n, 0, 
                                      (struct sockaddr*)&op->out.sendto_v.addr, 
                                      op->out.sendto_v.addrlen);
            result = (n == -1) ? errno : 0;
            bytes = (n == -1) ? 0 : n;
            if (op->callback.onsendto) {
                op->callback.onsendto(op->param, result, bytes, 
                                      op->out.sendto_v.addr, op->out.sendto_v.addrlen);
            }
            break;
        }
        
    case AIO_OP_RECVMSG: {
            struct sockaddr_storage peer, local;
            socklen_t peerlen = sizeof(peer), locallen = sizeof(local);
            socket_bufvec_t vec = {(void*)op->in.recvmsg.buffer, op->in.recvmsg.bytes};
            ssize_t n = socket_recvmsg_v(op->fd, &vec, 1, 0, 
                                        (struct sockaddr*)&peer, &peerlen, 
                                        (struct sockaddr*)&local, &locallen);
            result = (n == -1) ? errno : 0;
            bytes = (n == -1) ? 0 : n;
            if (op->callback.onrecvmsg) {
                op->callback.onrecvmsg(op->param, result, bytes, 
                                     &peer, peerlen, &local, locallen);
            }
            break;
        }
        
    case AIO_OP_SENDMSG: {
            socket_bufvec_t vec = {(void*)op->out.sendmsg.buffer, op->out.sendmsg.bytes};
            ssize_t n = socket_sendmsg_v(op->fd, &vec, 1, 0, 
                                        op->out.sendmsg.peer, op->out.sendmsg.peerlen, 
                                        op->out.sendmsg.local, op->out.sendmsg.locallen);
            result = (n == -1) ? errno : 0;
            bytes = (n == -1) ? 0 : n;
            if (op->callback.onsend) {
                op->callback.onsend(op->param, result, bytes);
            }
            break;
        }
        
    case AIO_OP_RECVMSG_V: {
            struct sockaddr_storage peer, local;
            socklen_t peerlen = sizeof(peer), locallen = sizeof(local);
            ssize_t n = socket_recvmsg_v(op->fd, op->in.recvmsg_v.vec, op->in.recvmsg_v.n, 0, 
                                        (struct sockaddr*)&peer, &peerlen, 
                                        (struct sockaddr*)&local, &locallen);
            result = (n == -1) ? errno : 0;
            bytes = (n == -1) ? 0 : n;
            if (op->callback.onrecvmsg) {
                op->callback.onrecvmsg(op->param, result, bytes, 
                                     &peer, peerlen, &local, locallen);
            }
            break;
        }
        
    case AIO_OP_SENDMSG_V: {
            ssize_t n = socket_sendmsg_v(op->fd, op->out.sendmsg_v.vec, op->out.sendmsg_v.n, 0, 
                                        op->out.sendmsg_v.peer, op->out.sendmsg_v.peerlen, 
                                        op->out.sendmsg_v.local, op->out.sendmsg_v.locallen);
            result = (n == -1) ? errno : 0;
            bytes = (n == -1) ? 0 : n;
            if (op->callback.onsend) {
                op->callback.onsend(op->param, result, bytes);
            }
            break;
        }
    }
}

// 提交AIO操作
static int aco_socket_submit(aco_sockopt_t* op) 
{
    aco_scheduler_t* sched = aco_worker_context();
    if (!sched) {
        aco_err("Worker is invalid! \n");
        op_pool_free(op);
        return -1;
    }

    // 检查socket有效性
    if (op->fd < 0) {
        aco_err("Invalid socket fd: %d \n", op->fd);
        op_pool_free(op);
        return -1;
    }
    
    aco_task_t* task = aco_task_pool_get(sched);
    if (!task) {
        aco_err("task get failed! \n");
        op_pool_free(op);
        return -1;
    }

    // 配置任务
    task->entry = aco_socket_entry;
    task->arg = op;
    task->state = ACO_TASK_WAITING;
    task->wait_type = ACO_WAIT_IO;
    task->wait_resource.io.fd = op->fd;
    task->wait_resource.io.events = op->events;
    task->wait_resource.io.revents = 0; // clean it.
    task->wait_resource.io.timeout_ms = op->timeout_ms;

    // 选择目标worker
    size_t target_id = aco_work_steal_min(sched);
    aco_worker_t* target = sched->worker_ctxs[target_id];
    task->worker_id = target_id;

    // 注册epoll事件 - 先MOD后ADD
    if (target->io_ctx) 
    {
        worker_io_ctx_t* io_ctx = (worker_io_ctx_t*)target->io_ctx;
        struct epoll_event ev;
        ev.events = op->events;
        ev.data.ptr = task;
        
        // 先尝试修改，如果失败再添加
        if (epoll_ctl(io_ctx->epoll_fd, EPOLL_CTL_MOD, op->fd, &ev) == -1) 
        {
            if (errno == ENOENT) { // 不存在，则添加
                if (epoll_ctl(io_ctx->epoll_fd, EPOLL_CTL_ADD, op->fd, &ev) == -1) 
                {
                    aco_err("epoll_ctl add failed for fd %d: %s\n", op->fd, strerror(errno));
                    op_pool_free(op);
                    aco_task_pool_put(sched, task);
                    return -1;
                }
            } 
            else {
                aco_err("epoll_ctl mod failed for fd %d: %s\n", op->fd, strerror(errno));
                op_pool_free(op);
                aco_task_pool_put(sched, task);
                return -1;
            }
        }
    }
    
    // 添加超时定时器
    if (op->timeout_ms > 0) 
    {
        aco_timer_del(task);
        task->wait_resource.timer.expire = system_clock() + op->timeout_ms;
        task->wait_resource.timer.interval = 0;
        aco_timer_add(task);
    }
    
    // 添加到等待队列
    aco_task_queue_push(&target->waiting_queue, task);
    aco_io_notify_worker(sched, target_id);
    
    return 0;
}

// ========== AIO操作接口 ==========
int aco_socket_accept(aco_socket_t socket, aco_onaccept onaccept, void* param) 
{
    aco_sockopt_t* op = (aco_sockopt_t*)socket;
    if (!op) return -1;
    
    op->type = AIO_OP_ACCEPT;
    op->param = param;
    op->callback.onaccept = onaccept;
    op->events = ACO_IO_RDONLY;
    op->timeout_ms = 0;
    
    return aco_socket_submit(op);
}

int aco_socket_connect(aco_socket_t socket, const struct sockaddr_storage* addr, socklen_t addrlen, 
                              aco_onconnect onconnect, void* param)
{
    aco_sockopt_t* op = (aco_sockopt_t*)socket;
    if (!op) return -1;
    
    op->type = AIO_OP_CONNECT;
    op->param = param;
    op->callback.onconnect = onconnect;
    op->out.connect.addr = addr;
    op->out.connect.addrlen = addrlen;
    op->events = ACO_IO_WRONLY;
    op->timeout_ms = 0;
    
    return aco_socket_submit(op);
}

int aco_socket_recv(aco_socket_t socket, void* buffer, size_t bytes, 
                         int timeout_ms, aco_onrecv onrecv, void* param) 
{
    aco_sockopt_t* op = (aco_sockopt_t*)socket;
    if (!op) return -1;

    // 注册epoll事件
    op->type = AIO_OP_RECV;
    op->param = param;
    op->callback.onrecv = onrecv;
    op->in.recv.buffer = buffer;
    op->in.recv.bytes = bytes;
    op->events = ACO_IO_RDONLY;
    op->timeout_ms = timeout_ms;
    
    return aco_socket_submit(op);
}

int aco_socket_send(aco_socket_t socket, const void* buffer, size_t bytes, 
                          aco_onsend onsend, void* param) 
{
    aco_sockopt_t* op = (aco_sockopt_t*)socket;
    if (!op) return -1;
    
    op->type = AIO_OP_SEND;
    op->param = param;
    op->callback.onsend = onsend;
    op->out.send.buffer = buffer;
    op->out.send.bytes = bytes;
    op->events = ACO_IO_WRONLY;
    op->timeout_ms = 0;
    
    return aco_socket_submit(op);
}

int aco_socket_recv_v(aco_socket_t socket, socket_bufvec_t* vec, int n, 
                             aco_onrecv onrecv, void* param) 
{
    aco_sockopt_t* op = (aco_sockopt_t*)socket;
    if (!op) return -1;
    
    op->type = AIO_OP_RECV_V;
    op->param = param;
    op->callback.onrecv = onrecv;
    op->in.recv_v.vec = vec;
    op->in.recv_v.n = n;
    op->events = ACO_IO_RDONLY;
    op->timeout_ms = 0;
    
    return aco_socket_submit(op);
}

int aco_socket_send_v(aco_socket_t socket, socket_bufvec_t* vec, int n, 
                             aco_onsend onsend, void* param) 
{
    aco_sockopt_t* op = (aco_sockopt_t*)socket;
    if (!op) return -1;
    
    op->type = AIO_OP_SEND_V;
    op->param = param;
    op->callback.onsend = onsend;
    op->out.send_v.vec = vec;
    op->out.send_v.n = n;
    op->events = ACO_IO_WRONLY;
    op->timeout_ms = 0;
    
    return aco_socket_submit(op);
}

int aco_socket_recvfrom(aco_socket_t socket, void* buffer, size_t bytes, 
                                aco_onrecvfrom onrecvfrom, void* param) 
{
    aco_sockopt_t* op = (aco_sockopt_t*)socket;
    if (!op) return -1;
    
    op->type = AIO_OP_RECVFROM;
    op->param = param;
    op->callback.onrecvfrom = onrecvfrom;
    op->in.recvfrom.buffer = buffer;
    op->in.recvfrom.bytes = bytes;
    op->events = ACO_IO_RDONLY;
    op->timeout_ms = 0;
    
    return aco_socket_submit(op);
}

int aco_socket_sendto(aco_socket_t socket, const void* buffer, size_t bytes, 
                             const struct sockaddr_storage* addr, socklen_t addrlen, 
                             aco_onsendto onsendto, void* param) 
{
    aco_sockopt_t* op = (aco_sockopt_t*)socket;
    if (!op) return -1;
    
    op->type = AIO_OP_SENDTO;
    op->param = param;
    op->callback.onsendto = onsendto;
    op->out.sendto.addr = addr;
    op->out.sendto.addrlen = addrlen;
    op->out.sendto.buffer = buffer;
    op->out.sendto.bytes = bytes;
    op->events = ACO_IO_WRONLY;
    op->timeout_ms = 0;
    
    return aco_socket_submit(op);
}

int aco_socket_recvfrom_v(aco_socket_t socket, socket_bufvec_t* vec, int n, 
                                  aco_onrecvfrom onrecvfrom, void* param) 
{
    aco_sockopt_t* op = (aco_sockopt_t*)socket;
    if (!op) return -1;
    
    op->type = AIO_OP_RECVFROM_V;
    op->param = param;
    op->callback.onrecvfrom = onrecvfrom;
    op->in.recvfrom_v.vec = vec;
    op->in.recvfrom_v.n = n;
    op->events = ACO_IO_RDONLY;
    op->timeout_ms = 0;
    
    return aco_socket_submit(op);
}

int aco_socket_sendto_v(aco_socket_t socket, socket_bufvec_t* vec, int n, 
                               const struct sockaddr_storage* addr, socklen_t addrlen,
                               aco_onsendto onsendto, void* param)
{
    aco_sockopt_t* op = (aco_sockopt_t*)socket;
    if (!op) return -1;
    
    op->type = AIO_OP_SENDTO_V;
    op->param = param;
    op->callback.onsendto = onsendto;
    op->out.sendto_v.addr = addr;
    op->out.sendto_v.addrlen = addrlen;
    op->out.sendto_v.vec = vec;
    op->out.sendto_v.n = n;
    op->events = ACO_IO_WRONLY;
    op->timeout_ms = 0;
    
    return aco_socket_submit(op);
}

int aco_socket_recvmsg(aco_socket_t socket, void* buffer, size_t bytes, 
                              aco_onrecvmsg onrecvmsg, void* param) 
{
    aco_sockopt_t* op = (aco_sockopt_t*)socket;
    if (!op) return -1;
    
    op->type = AIO_OP_RECVMSG;
    op->param = param;
    op->callback.onrecvmsg = onrecvmsg;
    op->in.recvmsg.buffer = buffer;
    op->in.recvmsg.bytes = bytes;
    op->events = ACO_IO_RDONLY;
    op->timeout_ms = 0;
    
    return aco_socket_submit(op);
}

int aco_socket_sendmsg(aco_socket_t socket, const void* buffer, size_t bytes, 
                              const struct sockaddr_storage* peer, socklen_t peerlen,
                              const struct sockaddr_storage* local, socklen_t locallen,
                              aco_onsend onsend, void* param)
{
    aco_sockopt_t* op = (aco_sockopt_t*)socket;
    if (!op) return -1;
    
    op->type = AIO_OP_SENDMSG;
    op->param = param;
    op->callback.onsend = onsend;
    op->out.sendmsg.peer = peer;
    op->out.sendmsg.peerlen = peerlen;
    op->out.sendmsg.local = local;
    op->out.sendmsg.locallen = locallen;
    op->out.sendmsg.buffer = buffer;
    op->out.sendmsg.bytes = bytes;
    op->events = ACO_IO_WRONLY;
    op->timeout_ms = 0;
    
    return aco_socket_submit(op);
}

int aco_socket_recvmsg_v(aco_socket_t socket, socket_bufvec_t* vec, int n, 
                                aco_onrecvmsg onrecvmsg, void* param) 
{
    aco_sockopt_t* op = (aco_sockopt_t*)socket;
    if (!op) return -1;
    
    op->type = AIO_OP_RECVMSG_V;
    op->param = param;
    op->callback.onrecvmsg = onrecvmsg;
    op->in.recvmsg_v.vec = vec;
    op->in.recvmsg_v.n = n;
    op->events = ACO_IO_RDONLY;
    op->timeout_ms = 0;
    
    return aco_socket_submit(op);
}

int aco_socket_sendmsg_v(aco_socket_t socket, socket_bufvec_t* vec, int n, 
                                const struct sockaddr_storage* peer, socklen_t peerlen,
                                const struct sockaddr_storage* local, socklen_t locallen,
                                aco_onsend onsend, void* param)
{
    aco_sockopt_t* op = (aco_sockopt_t*)socket;
    if (!op) return -1;
    
    op->type = AIO_OP_SENDMSG_V;
    op->param = param;
    op->callback.onsend = onsend;
    op->out.sendmsg_v.peer = peer;
    op->out.sendmsg_v.peerlen = peerlen;
    op->out.sendmsg_v.local = local;
    op->out.sendmsg_v.locallen = locallen;
    op->out.sendmsg_v.vec = vec;
    op->out.sendmsg_v.n = n;
    op->events = ACO_IO_WRONLY;
    op->timeout_ms = 0;
    
    return aco_socket_submit(op);
}

int aco_socket_process(int timeout) 
{
    aco_scheduler_t* sched = aco_worker_context();
    if (!sched) return -1;
    
    // TODO: maybe call aco_scheduler_run or other
    // DO nothing now.
    return 0;
}

